"""
要求：
1、使用Spark读取 orders.txt 文件
2、各城市销售额从大到小排序
3、全部城市，哪些商品类别在售卖
4、北京市有哪些商品类别在售卖
"""

from pyspark import SparkConf, SparkContext
import os
import json
os.environ['PYSPARK_PYTHON'] = "C:/Users/Lenovo/AppData/Local/Programs/Python/Python310/python.exe"

conf = SparkConf().setMaster("local[*]").setAppName("test_spark")
sc = SparkContext(conf=conf)

# TODO 需求1：城市销售额排序
# 1.1 读取文件得到RDD
file_rdd = sc.textFile("E:/excel学习文档-表格/黑马-spark数据/orders.txt")

# 1.2 取出一个个json字符串
json_str = file_rdd.flatMap(lambda x: x.split("|"))

# 1.3 将一个个json字符串转换为字典
json_dict = json_str.map(lambda x: json.loads(x))
# print(json_dict.collect())

# 1.4 取出城市和销售额数据
city_money = json_dict.map(lambda x: (x['areaName'], float(x['money'])))
# print(city_money.collect())

# 1.5 按城市分组，按销售额聚合
sum_money = city_money.reduceByKey(lambda x, y: x + y)
# print(sum_money.collect())

# 1.6 按销售额聚合结果进行排序
money_sort = sum_money.sortBy(lambda x: x[1], ascending=False, numPartitions=1)
# print(money_sort.collect())

# TODO 需求2：全部城市有哪些商品类别在售卖
# 2.1 取出全部的商品类别
cate = json_dict.map(lambda x: (x['category']))
# print(cate.collect())

# 2.2 对全部商品类别进行去重
cate_list = cate.distinct()
# print(cate_list.collect())

# TODO 需求3：北京市有哪些商品类别在售卖
# 3.1 过滤北京市的数据
beijing = json_dict.filter(lambda x: x['areaName'] == "北京")
# print(beijing.collect())

# 3.2 取出全部商品类别并去重
# print(beijing.map(lambda x: x['category']).distinct().collect())